package wsz.reconsume;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerOrderly;
import com.alibaba.rocketmq.common.message.MessageExt;

/**
 * 有序消费
 * @author wsz
 * @date 2022/2/25 13:14
 **/
public class OrderConsumer {
    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_grp_wsz");
        consumer.setNamesrvAddr("192.168.174.138:9876");
        // 每次从message queues中各取一条消息
        consumer.setConsumeMessageBatchMaxSize(1);
        // 一个线程消费
        consumer.setConsumeThreadMin(1);
        consumer.setConsumeThreadMax(1);
        // 主题订阅
        consumer.subscribe("tp_demo_05", "*");

        /**
         * 顺序消费
         * 消费失败进行重试，间隔1s
         */
        consumer.setMessageListener((MessageListenerOrderly) (msgs, context) -> {
            for (MessageExt msg: msgs) {
                System.out.println(msg.getMsgId() + "\t" +
                        msg.getQueueId() + "\t" +
                        new String(msg.getBody()));
            }
            // 触发重试
            return null;
        });

        consumer.start();
    }
}
